package cn.doitedu.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Demo13_UnnestArray {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        /**
         * id,name,scores
         * 1,zs,[80,90,60]
         * 2,ls,[88,98,80]
         * 3,ww,[66,77,55]
         */
        tenv.executeSql(
                "create temporary view stu_info as  " +
                "select 1 as id, 'zs' as name, array[80,90,60] as scores " +
                "union all " +
                "select 2 , 'ls', array[88,98,80] " +
                "union all " +
                "select 3 , 'ww', array[66,77,55] ");

        tenv.executeSql(
                "select id,name,score  " +
                "from  " +
                "stu_info " +
                "cross join " +
                "unnest(scores) as t (score)").print();


    }
}
